package thread;

import java.io.IOException;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import buffer.SocketBufferHandler;
import cache.LimitLatch;
import cache.SynchronizedStack;
import cachedata.NioChannel;
import config.SocketProperties;
import config.SystemConfig;
import util.ExceptionUtils;

/**
 * 监听socket链接请求
 * 
 * @author koukaiqiang
 *
 */
class Acceptor implements Runnable {

	private static final Logger log = LoggerFactory.getLogger(Acceptor.class);

	private static final int INITIAL_ERROR_DELAY = 50;

	private static final int MAX_ERROR_DELAY = 1600;

	private volatile boolean running = true;

	private ServerSocketChannel serverSock = null;

	private String threadName;

	private Poller[] pollers;

	private AtomicInteger pollerRotater = new AtomicInteger(0);

	private static LimitLatch connectionLimitLatch = new LimitLatch(SystemConfig.maxWait);

	/**
	 * Bytebuffer cache, each channel holds a set of buffers (two, except for
	 * SSL holds four)
	 */
	private SynchronizedStack<NioChannel> nioChannels;

	public Acceptor(ServerSocketChannel serverSock, String threadName, Poller[] pollers,
			SynchronizedStack<NioChannel> nioChannels) {
		super();
		this.serverSock = serverSock;
		this.threadName = threadName;
		this.pollers = pollers;
		this.nioChannels = nioChannels;
	}

	public void run() {

		int errorDelay = 0;
		while (running) {

			if (!running) {
				break;
			}

			try {
				// if we have reached max connections, wait
				countUpOrAwaitConnection();

				SocketChannel socket = null;
				try {
					// Accept the next incoming connection from the server
					// socket
					socket = serverSock.accept();
					log.debug("get a client accept");
				} catch (IOException ioe) {
					// we didn't get a socket
					countDownConnection();
					// Introduce delay if necessary
					errorDelay = handleExceptionWithDelay(errorDelay);
					// re-throw
					throw ioe;
				}
				// Successful accept, reset the error delay
				errorDelay = 0;

				// setSocketOptions() will add channel to the poller
				// if successful
				if (running) {
					if (!setSocketOptions(socket)) {
						countDownConnection();
						closeSocket(socket);
					}
				} else {
					countDownConnection();
					closeSocket(socket);
				}
			} catch (SocketTimeoutException sx) {
				// Ignore: Normal condition
			} catch (IOException x) {
				if (running) {
					// log.error(sm.getString("endpoint.accept.fail"), x);
					log.error(x.getMessage());
				}
			} catch (Throwable t) {
				ExceptionUtils.handleThrowable(t);
				// log.error(sm.getString("endpoint.accept.fail"), t);
				log.error(t.getMessage());
			}
		}
	}

	private boolean setSocketOptions(SocketChannel socket) {
		// Process the connection
		try {
			// disable blocking, APR style, we are gonna be polling it
			socket.configureBlocking(false);
			Socket sock = socket.socket();
			SocketProperties.getInstance().setProperties(sock);

			NioChannel channel = nioChannels.pop();
			if (channel == null) {
				SocketBufferHandler bufhandler = new SocketBufferHandler(
						SocketProperties.getInstance().getAppReadBufSize(),
						SocketProperties.getInstance().getAppWriteBufSize(),
						SocketProperties.getInstance().getDirectBuffer());
				channel = new NioChannel(socket, bufhandler);
			} else {
				channel.setIOChannel(socket);
				channel.reset();
			}
			log.debug(" new NioChannel ");
			getPoller().register(channel);
		} catch (Throwable t) {
			ExceptionUtils.handleThrowable(t);
			try {
				log.error("", t);
			} catch (Throwable tt) {
				ExceptionUtils.handleThrowable(tt);
			}
			// Tell to close the socket
			return false;
		}
		return true;
	}

	private Poller getPoller() {
		int idx = Math.abs(pollerRotater.incrementAndGet()) % pollers.length;
		return pollers[idx];
	}

	private void closeSocket(SocketChannel socket) {
		try {
			socket.socket().close();
		} catch (IOException ioe) {
			log.debug("", ioe);
		}
		try {
			socket.close();
		} catch (IOException ioe) {
			log.debug("", ioe);
		}

	}

	@SuppressWarnings("unused")
	public static void countUpOrAwaitConnection() throws InterruptedException {
		if (SystemConfig.maxConnections == -1)
			return;
		LimitLatch latch = connectionLimitLatch;
		if (latch != null)
			latch.countUpOrAwait();
	}

	@SuppressWarnings("unused")
	public static long countDownConnection() {
		if (SystemConfig.maxConnections == -1)
			return -1;
		LimitLatch latch = connectionLimitLatch;
		if (latch != null) {
			long result = latch.countDown();
			if (result < 0) {
				// getLog().warn(sm.getString("endpoint.warn.incorrectConnectionCount"));
			}
			return result;
		} else {
			return -1;
		}
	}

	protected int handleExceptionWithDelay(int currentErrorDelay) {
		// Don't delay on first exception
		if (currentErrorDelay > 0) {
			try {
				Thread.sleep(currentErrorDelay);
			} catch (InterruptedException e) {
				// Ignore
			}
		}

		// On subsequent exceptions, start the delay at 50ms, doubling the delay
		// on every subsequent exception until the delay reaches 1.6 seconds.
		if (currentErrorDelay == 0) {
			return INITIAL_ERROR_DELAY;
		} else if (currentErrorDelay < MAX_ERROR_DELAY) {
			return currentErrorDelay * 2;
		} else {
			return MAX_ERROR_DELAY;
		}
	}

	public final void setThreadName(final String threadName) {
		this.threadName = threadName;
	}

	public final String getThreadName() {
		return threadName;
	}
}